package com.holly.unit.kafka.config;

import lombok.ToString;

/**
 * kafka配置信息
 * @author pengpeng
 *
 */
@ToString
public class KafkaConfig {
	/**
	 * 这里改为你的kafka服务器ip和端口号 比如 192.168.1.1:19092,192.168.1.2:29092,192.168.1.3:39092
	 */
	public String bootstrapServers;
	/**
	 * 如果该值大于零时，表示启用重试失败的发送次数，默认为int的最大值
	 */
	public int    Retries = Integer.MAX_VALUE;
	/**
	 * 每当多个记录被发送到同一分区时，生产者将尝试将记录一起批量处理为更少的请求，默认值为16384(单位字节)16k
	 * 如果消息超过16K，则底层会分配消息的长度字节
	 */
	public int    batchSize = 16384;
	
	/**
	 * 生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为3355443 32M
	 */
	public int   bufferMemory = 33554432;
	
	/**
	 * key的序列化类 默认是String类型
	 */
	public String keySerializer = "org.apache.kafka.common.serialization.StringSerializer";
	
	/**
	 * value的序列化类 默认是String类型
	 */
	public String valueSerializer = "org.apache.kafka.common.serialization.StringSerializer";
	
	/**
	 * #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
      #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
      #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
      #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
      #可以设置的值为：all, 0, 1
	 */
	public String ack;
	
	/**
	 * 当Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了），该如何处理？
	 * earliest：自动重置偏移量到最早的偏移量。
	 * latest：默认，自动重置偏移量为最新的偏移量。
	 * none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
	 */
	public String  autoOffsetReset = "earliest";
	
	/**
	 * 如果'enable-auto-commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
	 */
	public String autoCommitInterval = "5000";
	
	/**
	 * key的反序列化器类 默认是String
	 */
	public String keyDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
	
	/**
	 * value的反序列化器类 默认是String
	 */
	public String valueDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
	
	/**
	 * 发送消息事务id,用户可以手动设定，默认为 tranactional_id_01
	 */
	public String tranactionalId = "tranactional_id_01";
	
	/**
	 * 如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。单位 ms，默认值是 0ms，表示没有
       延迟。生产环境建议该值大小为 5-100ms 之间
	 */
	public String lingerMs = "0";
	
	/**
	 * 连接池，最小连接数
	 */
	public int producerMinPoolSize = 3;
	
	/**
	 * 连接池，最大连接数
	 */
	public int producerMaxPoolSize = 8;
	
	/**
	 * 连接池，是否使用懒加载，默认true
	 */
	public Boolean isLazy = true;
	
	


	public String getTranactionalId() {
		return tranactionalId;
	}

	public void setTranactionalId(String tranactionalId) {
		this.tranactionalId = tranactionalId;
	}

	public String getBootstrapServers() {
		return bootstrapServers;
	}

	public void setBootstrapServers(String bootstrapServers) {
		this.bootstrapServers = bootstrapServers;
	}

	public int getRetries() {
		return Retries;
	}

	public void setRetries(int retries) {
		Retries = retries;
	}

	public int getBatchSize() {
		return batchSize;
	}

	public void setBatchSize(int batchSize) {
		this.batchSize = batchSize;
	}

	public int getBufferMemory() {
		return bufferMemory;
	}

	public void setBufferMemory(int bufferMemory) {
		this.bufferMemory = bufferMemory;
	}

	public String getKeySerializer() {
		return keySerializer;
	}

	public void setKeySerializer(String keySerializer) {
		this.keySerializer = keySerializer;
	}

	public String getValueSerializer() {
		return valueSerializer;
	}

	public void setValueSerializer(String valueSerializer) {
		this.valueSerializer = valueSerializer;
	}

	public String getAck() {
		return ack;
	}

	public void setAck(String ack) {
		this.ack = ack;
	}

	public String getAutoOffsetReset() {
		return autoOffsetReset;
	}

	public void setAutoOffsetReset(String autoOffsetReset) {
		this.autoOffsetReset = autoOffsetReset;
	}

	public String getAutoCommitInterval() {
		return autoCommitInterval;
	}

	public void setAutoCommitInterval(String autoCommitInterval) {
		this.autoCommitInterval = autoCommitInterval;
	}

	public String getKeyDeserializer() {
		return keyDeserializer;
	}

	public void setKeyDeserializer(String keyDeserializer) {
		this.keyDeserializer = keyDeserializer;
	}

	public String getValueDeserializer() {
		return valueDeserializer;
	}

	public void setValueDeserializer(String valueDeserializer) {
		this.valueDeserializer = valueDeserializer;
	}
	
}
